package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
)

const (
	rocketmqNamesrv = "127.0.0.1:9876" // RocketMQ NameServer 的地址
)

func main() {
	// 创建生产者实例
	p, err := rocketmq.NewProducer(
		producer.WithNameServer([]string{rocketmqNamesrv}),
	)
	if err != nil {
		log.Fatalf("Failed to create producer: %s\n", err)
	}

	// 启动生产者
	err = p.Start()
	if err != nil {
		log.Fatalf("Failed to start producer: %s\n", err)
	}
	defer p.Shutdown()

	// 创建消费者实例
	c, err := rocketmq.NewPushConsumer(
		consumer.WithGroupName("testGroup"),
		consumer.WithNameServer([]string{rocketmqNamesrv}),
	)
	if err != nil {
		log.Fatalf("Failed to create consumer: %s\n", err)
	}

	// 订阅消息
	err = c.Subscribe("testTopic", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for _, msg := range msgs {
			log.Printf("Received message: %s\n", msg.Body)
		}
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		log.Fatalf("Failed to subscribe: %s\n", err)
	}

	// 启动消费者
	err = c.Start()
	if err != nil {
		log.Fatalf("Failed to start consumer: %s\n", err)
	}
	defer c.Shutdown()

	// 发布消息
	for i := 1; i <= 3; i++ {
		msg := fmt.Sprintf("Hello RocketMQ! Message %d", i)
		result, err := p.SendSync(context.Background(), &primitive.Message{
			Topic: "testTopic",
			Body:  []byte(msg),
		})
		if err != nil {
			log.Printf("Failed to publish message: %s\n", err)
		} else {
			log.Printf("Message successfully published: %s\n", result.MsgID)
		}
	}

	// 等待信号（Ctrl+C）关闭程序
	signalChan := make(chan os.Signal, 1)
	signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
	<-signalChan
}
